package cn.gwm.flink.streaming.function.map;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

/**
 * 数据子编码转换
 * @author GW00256253
 */
@Slf4j
public class OriginalAllDataMapFunction implements FlatMapFunction<String, JSONObject> {

    private String vehicleType;

    /**
     * 接收字段集合列表
     */
    List<String> list;

    Map<String,String> map;

    public OriginalAllDataMapFunction(String vehicleType , List<String> list,Map<String,String> map){
        this.list = list;
        this.vehicleType = vehicleType;
        this.map = map;
    }


    @Override
    public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
        JSONObject object = null;
        try {
            object = JSONUtil.parseObj(s);
        } catch (Exception e) {
            log.error("源数据转json异常，源数据格式有误！");
            e.printStackTrace();
            throw e;
        }
        //周期can信号
        if(BaseFields.COMMAND_STATUS.equals(object.getStr(BaseFields.command))
            ||BaseFields.COMMAND_TBOX.equals(object.getStr(BaseFields.command))
            ||BaseFields.COMMAND_REMOTE_REQUEST.equals(object.getStr(BaseFields.command))
            ){
            JSONObject resultOjb = JSONUtil.createObj();
            for (Map.Entry<String,String> entry : map.entrySet()) {
                if(object.containsKey(entry.getKey())){
                    resultOjb.set(entry.getValue(), object.getStr(entry.getKey()));
                }
            }
            if(BaseFields.COMMAND_TBOX.equals(object.getStr(BaseFields.command))){
                try {
                    if (resultOjb.getDouble(BaseFields.LONGITUDE) > 72
                            && resultOjb.getDouble(BaseFields.LONGITUDE) < 136 && resultOjb.getDouble(BaseFields.LATITUDE) > 3
                            && resultOjb.getDouble(BaseFields.LATITUDE) < 54) {
                        resultOjb.set(BaseFields.IS_GPS_FLAG,object.getStr(BaseFields.IS_GPS_FLAG));
                        collectData(collector, resultOjb);
                    }
                } catch (Exception e) {
                    log.error("缩减字段异常！object={},stackTrace={}", object, e.getStackTrace());
                    e.printStackTrace();
                    throw e;
                }
            }else{
                if(BaseFields.COMMAND_REMOTE_REQUEST.equals(object.getStr(BaseFields.command))){
                    resultOjb.set(BaseFields.REQUEST_TIME, object.getStr(BaseFields.tid));
                }
                collectData(collector, resultOjb);
            }
        }
    }
    private void collectData(Collector<JSONObject> collector, JSONObject object) {
        try {
            object.set(BaseFields.STORE_TIME, System.currentTimeMillis())
                    .set(VehicleType.VEHICLE_TYPE, vehicleType)
                    .set(BaseFields.PARTITION_KEY, DateUtil.format(DateUtil.date(object.getLong(BaseFields.tid)),
                            DatePattern.NORM_DATE_PATTERN));
        } catch (Exception e) {
            log.error("json字段不存在item_time字段异常！object={},stackTrace={}", object.toString(), e.getStackTrace());
            e.printStackTrace();
            throw e;
        }
        collector.collect(object);

    }
}
